package com.lzwk.app.ods.adb;

/**
 * @Author: CC
 * @Date: 2022/1/5 17:57
 */

import com.alibaba.fastjson.JSONObject;
import com.lzwk.beans.ocpx.AdbClassStudent;
import com.lzwk.beans.ocpx.WkFormAccountInfo;
import com.lzwk.common.adb.ADBConstant;
import com.lzwk.common.wkform.WkFormConstant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;


public class AdbCDCTables {

    //本地测试port：11810

    public static  void adbclassStudent(StreamExecutionEnvironment env,StreamTableEnvironment tenv) {


        MySqlSource<String> mySqlSourceHistory1 = MySqlSource.<String>builder()
                .hostname(ADBConstant.ADBSLAVEHOST)
                .port(Integer.parseInt(ADBConstant.ADBSLAVEPORT))
                .username(ADBConstant.ADBSLAVEUSERNAME)
                .password(ADBConstant.ADBSLAVEPASSWORD)
                .databaseList(ADBConstant.ADBSLAVEDBNAME)
                .tableList(ADBConstant.ADBSLAVEDBNAME + "." + ADBConstant.ADBJDBCDIVISIONCLASSSTUDENT)
                .deserializer(new AdbCustomerDeserialization("withop"))
                .startupOptions(StartupOptions.initial())
                .includeSchemaChanges(true)
                .splitSize(80000)
                .build();

        SingleOutputStreamOperator<AdbClassStudent> source1 = env
                .fromSource(mySqlSourceHistory1,WatermarkStrategy
                                .noWatermarks()
                        , "MySQL adb Source")
                .process(new ProcessFunction<String, AdbClassStudent>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<AdbClassStudent> out) throws Exception {
                        AdbClassStudent adbClassStudent = null;
                        JSONObject data = JSONObject.parseObject(value);
                        if (data.getJSONObject("data").toJSONString() != null) {
                            String dataDetail = data.getJSONObject("data").toJSONString();
                            adbClassStudent = JSONObject.parseObject(dataDetail, AdbClassStudent.class);
                        }
                        out.collect(adbClassStudent);
                    }
                });


        Table adbStudentInfoTable = tenv.fromDataStream(source1,
                $("account_id"), $("term_class_id"), $("division_activity_id"), $("activity_term_id"), $("state"), $("refund_status"),
                $("remark"), $("create_time"),$("create_time").rowtime().as("watermark_time"), $("update_time"), $("op"), $("id"), $("pt").proctime());

        tenv.createTemporaryView("adb_division_class_student", adbStudentInfoTable);







/*        MySqlSource<String> mySqlSourceNew = MySqlSource.<String>builder()
                .hostname(ADBConstant.ADBSLAVEPORT)
                .port(Integer.parseInt(ADBConstant.ADBSLAVEPORT))
                .username(ADBConstant.ADBSLAVEUSERNAME)
                .password(ADBConstant.ADBSLAVEPASSWORD)
                .databaseList(ADBConstant.ADBSLAVEDBNAME)
                .tableList(ADBConstant.ADBSLAVEDBNAME + "." + ADBConstant.ADBJDBCDIVISIONCLASSSTUDENT)
                .deserializer(new AdbCustomerDeserialization("withop"))
                .startupOptions(StartupOptions.latest())
                .includeSchemaChanges(true)
                .build();
        SingleOutputStreamOperator<AdbClassStudent> source2 = env
                .fromSource(mySqlSourceNew, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .process(new ProcessFunction<String, AdbClassStudent>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<AdbClassStudent> out) throws Exception {
                        AdbClassStudent adbClassStudent = null;
                        JSONObject data = JSONObject.parseObject(value);
                        if (data.getJSONObject("data").toJSONString() != null) {
                            String dataDetail = data.getJSONObject("data").toJSONString();
                            adbClassStudent = JSONObject.parseObject(dataDetail, AdbClassStudent.class);
                        }
                        out.collect(adbClassStudent);
                    }
                });*/

        //Stream 不支持读取历史binlog的同时也读取最新的binlog变化，因此需要分别读取最后再union起来

/*        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        Table table = tenv.fromDataStream(source,
                $("account_id"), $("term_class_id"), $("division_activity_id"), $("activity_term_id"), $("state"), $("refund_status"),
                $("remark"), $("create_time").rowtime(), $("update_time"), $("id"), $("op"), $("pt").proctime());

        table.execute().print();*/

    }

    public static TableResult contentMapping(StreamTableEnvironment tenv){

        TableResult tableResult = tenv.executeSql("create table adb_content_mapping(" +
                " `id` INT, " +
                " `object_type` STRING, " +
                " `origin_object_id` INT, " +
                " `target_object_ids` STRING, " +
                " `map_type` STRING, " +
                " `origin_update_time` TIMESTAMP(0), " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + ADBConstant.ADBSLAVEDBNAME + "', " +
                "   'port'='" + ADBConstant.ADBSLAVEPORT + "', " +
                "   'username'='" + ADBConstant.ADBSLAVEUSERNAME + "', " +
                "   'password'='" + ADBConstant.ADBSLAVEPASSWORD + "', " +
                "   'database-name'='" + ADBConstant.ADBSLAVEDBNAME + "', " +
                "   'table-name'='" + ADBConstant.ADBCDCCONTENTMAPPING + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")");



        tenv.sqlQuery(" select * from adb_content_mapping ").execute().print();

        return tableResult;

    }






}
